package belf.migrate.api.source.cdc;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.queue.RedisQueue;
import belf.migrate.api.source.Boundedness;
import belf.migrate.api.source.ServerIdRange;
import belf.migrate.api.source.Source;
import belf.migrate.api.taskconf.ConnectionConf;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.taskconf.SourceConf;
import belf.migrate.api.util.EngineConf;
import belf.migrate.core.util.ConfigUtil;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 嵌入Debezium Engine以捕获数据库的ChangedEvent事件。定制化参数参考{@link belf.migrate.api.taskconf.TaskConf#getCdc()}}
 */
@Slf4j
public abstract class CDCSource implements Source {
    private DebeziumEngine<ChangeEvent<String, String>> engine;

    private final JobContext jobContext;

    private final SourceConf conf;

    private final ConnectionConf connectionConf;
    /** debezium Connector的定制参数 */
    private final Map<String, String> cdc;

    public CDCSource(JobContext jobContext) {
        this.jobContext = jobContext;
        conf = jobContext.getTaskConf().getSourceConf();
        connectionConf = conf.getConnectionConf();
        cdc = jobContext.getTaskConf().getCdc();
    }

    public JobContext getJobContext() {
        return jobContext;
    }

    @Override
    public Boundedness getBoundedness() {
        return Boundedness.UNBOUNDED;
    }

    private Properties config() {
        Properties props = new Properties();
        props.setProperty("name", "debezium-job-" + jobContext.getJobId());
        props.setProperty("database.server.id", "" + ServerIdRange.serverId());
        props.setProperty("database.server.name", conf.getCatalogName() + "-" + jobContext.getJobId());

        props.setProperty("database.hostname", connectionConf.getHost());
        props.setProperty("database.port", "" + connectionConf.getPort());
        props.setProperty("database.user", connectionConf.getUser());
        props.setProperty("database.password", connectionConf.getPassword());

        //offset config begin
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.storage.file.filename",
                ConfigUtil.getConfigDir() + File.separator + "cdc" + File.separator + jobContext.getJobId() + "_offsets.dat");

        //other default properties
        props.setProperty("topic.prefix", "belf-migrate-connector-" + jobContext.getJobId());
        props.setProperty("schema.history.internal",
                "io.debezium.storage.file.history.FileSchemaHistory");
        props.setProperty("schema.history.internal.file.filename",
                ConfigUtil.getConfigDir() + File.separator + "cdc" + File.separator + jobContext.getJobId() + "_history.dat");
        props.setProperty("database.history",
                "io.debezium.relational.history.FileDatabaseHistory");
        props.setProperty("database.history.file.filename",
                ConfigUtil.getConfigDir() + File.separator + "cdc" + File.separator + jobContext.getJobId() + "_dbhistory.log");
        props.setProperty("converter.schemas.enable", "false"); // don't include schema in message

        //WKT格式
        props.setProperty("db.conversion.geometry", "true");
        props.setProperty("db.conversion.geometry.type", "wkt");
        props.setProperty("decimal.handling.mode", "double");
        // datetime transformer
//        props.setProperty("transforms", "DateTimeField");
//        props.setProperty("transforms.DateTimeField.type", "org.apache.kafka.connect.transforms.TimestampConverter$Value");
//        props.setProperty("transforms.DateTimeField.field", "last_update");
//        props.setProperty("transforms.DateTimeField.target.type", "string");
//        props.setProperty("transforms.DateTimeField.format", "yyyy-MM-dd HH:mm:ss");

        props.setProperty("snapshot.mode", "always");

        for (Map.Entry<String, String> entry : cdc.entrySet()) {
            props.setProperty(entry.getKey(), entry.getValue());
        }

        return props;
    }

    /**
     * 启动debezium嵌入式Engine以捕获数据库的ChangedEvent，代码参考的debezium官方例子
     */
    @Override
    public void start() {
        Properties props = config();

        engine = DebeziumEngine.create(Json.class)
                .using(props)
                .notifying(record -> {
                    try {
                        if (record.value().contains("CUSTOMERS")) {
                            log.debug("debug here");
                        }
                        String json = CDCEventConverter.to(record.key(), record.value());
                        if (null != json) {
                            RedisQueue.push(jobContext.getJobId(), json);
                        }
                    } catch (Exception e) {
                        log.error("DDL event not support yet: {}", e.getMessage());
                    }
                })
                .using((success, message, error) -> {
                    if (error != null) {
                        // 报错回调
                        log.error("ChangedEvent error, message:" + message + "exception:" + error);
                        error.printStackTrace();
                    }
                    closeEngine(engine);
                })
                .build();

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(engine);
        addShutdownHook(engine);  //注册JVM hook，JVM退出时自动关闭debezium连接
        awaitTermination(executor);
    }

    @Override
    public void stop() {
        closeEngine(engine);
        try {
            jobContext.getSourceCatalog().close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void restore(JobContext newJobContext) {
        throw new UnsupportedOperationException(
                "restore(newJobContext) method has not been implemented.");
    }

    private static void closeEngine(DebeziumEngine<ChangeEvent<String, String>> engine) {
        try {
            engine.close();
        } catch (IOException ignored) {
        }
    }

    private static void addShutdownHook(DebeziumEngine<ChangeEvent<String, String>> engine) {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> closeEngine(engine)));
    }

    private static void awaitTermination(ExecutorService executor) {
        if (executor != null) {
            try {
                executor.shutdown();
                while (!executor.awaitTermination(
                        EngineConf.getInstance().getInt("cdc.await.terminate.seconds", 5),
                        TimeUnit.SECONDS)) {
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    @Override
    public void setJobContext(JobContext jobContext) {
        Source.super.setJobContext(jobContext);
    }
}
